package com.kis.core.scan.factory;

import com.caucho.hessian.HessianException;
import com.caucho.hessian.client.HessianConnectionException;
import com.caucho.hessian.client.HessianConnectionFactory;
import com.caucho.hessian.client.HessianProxyFactory;
import com.caucho.hessian.client.HessianRuntimeException;
import com.caucho.hessian.io.SerializerFactory;
import com.kis.core.util.HessianBalance;
import com.kis.core.util.Timer;
import com.kis.core.util.http.HttpUtils;
import com.kis.core.util.mac.MacUtils;
import com.kis.core.util.model.HessianBean;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.core.env.Environment;
import org.springframework.remoting.RemoteAccessException;
import org.springframework.remoting.RemoteConnectFailureException;
import org.springframework.remoting.RemoteLookupFailureException;
import org.springframework.remoting.RemoteProxyFailureException;
import org.springframework.util.Assert;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.UndeclaredThrowableException;
import java.net.ConnectException;
import java.net.MalformedURLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by wongloong on 16-4-25.
 */
public class LoadBalanceHessianClientInterceptor extends LoadBalanceUrlBasedRemoteAccessor implements MethodInterceptor {
    private HessianProxyFactory proxyFactory = new HessianProxyFactory();
    private List<HessianBean> hessianBeanList = Collections.synchronizedList(new ArrayList<HessianBean>());
    private AtomicBoolean regis = new AtomicBoolean(false);
    @Autowired
    Environment env;
    @Autowired
    DiscoveryClient discoveryClient;

    /**
     * Set the HessianProxyFactory instance to use.
     * If not specified, a default HessianProxyFactory will be created.
     * <p>Allows to use an externally configured factory instance,
     * in particular a custom HessianProxyFactory subclass.
     */
    public void setProxyFactory(HessianProxyFactory proxyFactory) {
        this.proxyFactory = (proxyFactory != null ? proxyFactory : new HessianProxyFactory());
    }

    /**
     * Specify the Hessian SerializerFactory to use.
     * <p>This will typically be passed in as an inner bean definition
     * of type {@code com.caucho.hessian.io.SerializerFactory},
     * with custom bean property values applied.
     */
    public void setSerializerFactory(SerializerFactory serializerFactory) {
        this.proxyFactory.setSerializerFactory(serializerFactory);
    }

    /**
     * Set whether to send the Java collection type for each serialized
     * collection. Default is "true".
     */
    public void setSendCollectionType(boolean sendCollectionType) {
        this.proxyFactory.getSerializerFactory().setSendCollectionType(sendCollectionType);
    }

    /**
     * Set whether to allow non-serializable types as Hessian arguments
     * and return values. Default is "true".
     */
    public void setAllowNonSerializable(boolean allowNonSerializable) {
        this.proxyFactory.getSerializerFactory().setAllowNonSerializable(allowNonSerializable);
    }

    /**
     * Set whether overloaded methods should be enabled for remote invocations.
     * Default is "false".
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setOverloadEnabled
     */
    public void setOverloadEnabled(boolean overloadEnabled) {
        this.proxyFactory.setOverloadEnabled(overloadEnabled);
    }

    /**
     * Set the username that this factory should use to access the remote service.
     * Default is none.
     * <p>The username will be sent by Hessian via HTTP Basic Authentication.
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setUser
     */
    public void setUsername(String username) {
        this.proxyFactory.setUser(username);
    }

    /**
     * Set the password that this factory should use to access the remote service.
     * Default is none.
     * <p>The password will be sent by Hessian via HTTP Basic Authentication.
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setPassword
     */
    public void setPassword(String password) {
        this.proxyFactory.setPassword(password);
    }

    /**
     * Set whether Hessian's debug mode should be enabled.
     * Default is "false".
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setDebug
     */
    public void setDebug(boolean debug) {
        this.proxyFactory.setDebug(debug);
    }

    /**
     * Set whether to use a chunked post for sending a Hessian request.
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setChunkedPost
     */
    public void setChunkedPost(boolean chunkedPost) {
        this.proxyFactory.setChunkedPost(chunkedPost);
    }

    /**
     * Specify a custom HessianConnectionFactory to use for the Hessian client.
     */
    public void setConnectionFactory(HessianConnectionFactory connectionFactory) {
        this.proxyFactory.setConnectionFactory(connectionFactory);
    }

    /**
     * Set the socket connect timeout to use for the Hessian client.
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setConnectTimeout
     */
    public void setConnectTimeout(long timeout) {
        this.proxyFactory.setConnectTimeout(timeout);
    }

    /**
     * Set the timeout to use when waiting for a reply from the Hessian service.
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setReadTimeout
     */
    public void setReadTimeout(long timeout) {
        this.proxyFactory.setReadTimeout(timeout);
    }

    /**
     * Set whether version 2 of the Hessian protocol should be used for
     * parsing requests and replies. Default is "false".
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setHessian2Request
     */
    public void setHessian2(boolean hessian2) {
        this.proxyFactory.setHessian2Request(hessian2);
        this.proxyFactory.setHessian2Reply(hessian2);
    }

    /**
     * Set whether version 2 of the Hessian protocol should be used for
     * parsing requests. Default is "false".
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setHessian2Request
     */
    public void setHessian2Request(boolean hessian2) {
        this.proxyFactory.setHessian2Request(hessian2);
    }

    /**
     * Set whether version 2 of the Hessian protocol should be used for
     * parsing replies. Default is "false".
     *
     * @see com.caucho.hessian.client.HessianProxyFactory#setHessian2Reply
     */
    public void setHessian2Reply(boolean hessian2) {
        this.proxyFactory.setHessian2Reply(hessian2);
    }


    @Override
    public void afterPropertiesSet() {
        super.afterPropertiesSet();
        prepare();
    }

    /**
     * Initialize the Hessian proxy for this interceptor.
     *
     * @throws RemoteLookupFailureException if the service name is invalid
     */
    public void prepare() throws RemoteLookupFailureException {
        checkHasRegis();
        try {
            if (regis.get()) {
                createHessianProxy4LoadBalance(this.proxyFactory);
            } else {
                System.exit(-1);
            }
        } catch (MalformedURLException ex) {
            throw new RemoteLookupFailureException("Service URL [" + getServerName() + "] is invalid", ex);
        }
    }

    /**
     * create hessian proxy list
     *
     * @param proxyFactory
     * @return List of hessian proxy
     * @throws MalformedURLException
     */
    protected void createHessianProxy4LoadBalance(HessianProxyFactory proxyFactory) throws MalformedURLException {
        Assert.notNull(getServiceInterface(), "'serviceInterface' is required");
        List<ServiceInstance> serviceInstances = discoveryClient.getInstances(getServerName().toLowerCase());
        for (ServiceInstance serviceInstance : serviceInstances) {
            Object hessianService = proxyFactory.create(getServiceInterface(), serviceInstance.getUri() + "/" + getAccessPath(), getBeanClassLoader());
            HessianBean hessianBean = new HessianBean();
            hessianBean.setServerName(getServerName().toLowerCase());
            hessianBean.setAccessPath(getAccessPath());
            hessianBean.setPort(serviceInstance.getPort());
            hessianBean.setHessian(hessianService);
            hessianBean.setUri(serviceInstance.getUri().toString());
            hessianBean.setHessianInterface(getServiceInterface());
            hessianBeanList.add(hessianBean);
        }
        Timer.getInstance(discoveryClient, hessianBeanList, proxyFactory);
    }

    private AtomicInteger retry = new AtomicInteger(0);

    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
//        if (this.hessianBeanList.isEmpty()) {
//            throw new IllegalStateException("HessianClientInterceptor is not properly initialized - " +
//                    "invoke 'prepare' before attempting any operations");
//        }
        Object theBestHessianProxy = null;
        ClassLoader originalClassLoader = overrideThreadContextClassLoader();
        try {
            theBestHessianProxy = HessianBalance.CheckBestHessianProxy(getAccessPath());
            if (null == theBestHessianProxy) {
                throw new HessianException("HessianClientInterceptor is not properly initialized - " +
                        "invoke 'prepare' before attempting any operations");
            }
            return invocation.getMethod().invoke(theBestHessianProxy, invocation.getArguments());
        } catch (InvocationTargetException ex) {
            Throwable targetEx = ex.getTargetException();
            // Hessian 4.0 check: another layer of InvocationTargetException.
            if (targetEx instanceof InvocationTargetException) {
                targetEx = ((InvocationTargetException) targetEx).getTargetException();
            }
            if (targetEx instanceof HessianConnectionException) {
                throw convertHessianAccessException(targetEx);
            } else if (targetEx instanceof HessianException || targetEx instanceof HessianRuntimeException) {
                Throwable cause = targetEx.getCause();
                if (cause.toString().contains("java.net.ConnectException: Connection refused")) {
                    //主动通知更新缓存
                    CheckHessianProxyNow(theBestHessianProxy);
                }
                throw convertHessianAccessException(cause != null ? cause : targetEx);
            } else if (targetEx instanceof UndeclaredThrowableException) {
                UndeclaredThrowableException utex = (UndeclaredThrowableException) targetEx;
                throw convertHessianAccessException(utex.getUndeclaredThrowable());
            } else {
                throw targetEx;
            }
        } catch (Throwable ex) {
            throw new RemoteProxyFailureException(
                    "Failed to invoke Hessian proxy for remote service [" + getServerName() + "]", ex);
        } finally {
            resetThreadContextClassLoader(originalClassLoader);
        }
    }

    /**
     * Convert the given Hessian access exception to an appropriate
     * Spring RemoteAccessException.
     *
     * @param ex the exception to convert
     * @return the RemoteAccessException to throw
     */
    protected RemoteAccessException convertHessianAccessException(Throwable ex) {
        if (ex instanceof HessianConnectionException || ex instanceof ConnectException) {
            return new RemoteConnectFailureException(
                    "Cannot connect to Hessian remote service at [" + getServerName() + "]", ex);
        } else {
            return new RemoteAccessException(
                    "Cannot access Hessian remote service at [" + getServerName() + "]", ex);
        }
    }

    public void CheckHessianProxyNow(Object problemHessianProxy) {
        List<ServiceInstance> serviceInstances = discoveryClient.getInstances(getServerName().toLowerCase());
        hessianBeanList.clear();
        for (ServiceInstance serviceInstance : serviceInstances) {
            Object hessianService = null;
            try {
                hessianService = proxyFactory.create(getServiceInterface(), serviceInstance.getUri() + "/" + getAccessPath(), getBeanClassLoader());
                HessianBean hessianBean = new HessianBean();
                hessianBean.setServerName(getServerName().toLowerCase());
                hessianBean.setAccessPath(getAccessPath());
                hessianBean.setPort(serviceInstance.getPort());
                hessianBean.setHessian(hessianService);
                hessianBean.setUri(serviceInstance.getUri().toString());
                hessianBean.setHessianInterface(getServiceInterface());
                if (hessianService.equals(problemHessianProxy)) {
                    HessianBalance.setHessiansWithKey(hessianBean.getPort(), getAccessPath());
                    continue;
                } else {
                    hessianBeanList.add(hessianBean);
                }
            } catch (MalformedURLException e) {
                e.printStackTrace();
            }
        }
    }

    public void checkHasRegis() {
        String macs = env.getProperty("macs");
        String authRequired = env.getProperty("auth_required");
        if (null != authRequired && null != macs && Boolean.parseBoolean(authRequired)) {
            String localMac = MacUtils.getMac();
            if (macs.indexOf(",") > -1) {
                String[] macArray = macs.split(",");
                for (String m : macArray) {
                    if (m.equalsIgnoreCase(localMac) && HttpUtils.sendServerRunInfo()) {
                        regis = new AtomicBoolean(true);
                        break;
                    }
                }
            } else if (macs.equalsIgnoreCase(localMac) && HttpUtils.sendServerRunInfo()) {
                regis = new AtomicBoolean(true);
            }
        } else if (null != authRequired && !Boolean.parseBoolean(authRequired)) {
            regis = new AtomicBoolean(true);
        }
    }
}
